package exams.concurrent;

import java.sql.Array;
import java.util.*;
import java.util.concurrent.*;

import static java.lang.Thread.sleep;

public class RunnableSolutionTest {

    public static void main(String[] args) {
        String[] datas = {"11", "14", "1", "2", "3", "4", "5", "6", "7"};

//        Map<Integer, ExecutorService> threadMap = new ConcurrentHashMap<>();
//        for(String partitionKey: datas) {
//            int threadNum = Integer.parseInt(partitionKey) % 10;
//            if(!threadMap.containsKey(threadNum)) {
//                threadMap.put(threadNum, Executors.newSingleThreadExecutor());
//            }
//            threadMap.get(threadNum).execute(() -> {
//                mockBlockProcess(partitionKey);
//            });
//        }
//
//        threadMap.entrySet().forEach(entry -> entry.getValue().shutdown());

        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Map<Integer, ArrayDeque<String>> map2 = new ConcurrentHashMap<>();
        for(String partitionKey: datas) {
            int pN = Integer.parseInt(partitionKey) % 2;
            if(!map2.containsKey(pN)) {
                map2.put(pN, new ArrayDeque<>());
            }
            map2.get(pN).add(partitionKey);
        }

        for(Map.Entry<Integer, ArrayDeque<String>> entry: map2.entrySet()) {
            executorService.execute(()-> {
                while (!entry.getValue().isEmpty()) {
                    String cur =  entry.getValue().poll();
                    mockBlockProcess(cur, entry.getKey());
                }
            });
        }

        executorService.shutdown();
    }

    static void mockBlockProcess(String str, int pN) {
        try {
            sleep(Integer.parseInt(str) * 1000);
            System.out.println(Thread.currentThread().getName() + " " + pN + " out " + str);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }



}
